package com.hotutil.demo;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdfs.DFSInotifyEventInputStream;
import org.apache.hadoop.hdfs.client.HdfsAdmin;
import org.apache.hadoop.hdfs.inotify.Event;
import org.apache.hadoop.hdfs.inotify.EventBatch;
import org.apache.hadoop.hdfs.inotify.MissingEventsException;

import java.io.IOException;
import java.net.URI;
import java.text.SimpleDateFormat;


public class HDFSNotifyDemo {
    //这里用static声明两个变量，因为static声明后的变量被称为gc roots， GC不会主动回收gc roots的变量。用static避免内存泄露
    public static  HdfsAdmin admin =null;
    public static DFSInotifyEventInputStream inotifyEventStream=null;
    /**
     * 解析edit文件信息的方法，通过此方法监控Test02目录
     */
    public static void TestEvent() throws IOException, MissingEventsException, InterruptedException {
         admin = new HdfsAdmin(URI.create("hdfs://10.7.221.29:8020"), new Configuration());
        /*
            getInotifyEventStream（），获取事件流的对象
            注意：此方法必须是单线程,不能在多个线程之间共享流对象。
            获取流的方式：当一个文件被DN读取后，会生成一个MetadataUpdateEvent对象并发送到NN，如果发生修改（追加、删除、移动、复制），attime会更新，再次发送MetadataUpdateEvent
            不需要我们手动获取，这里会自动接收DN推送的Event.
         */
         inotifyEventStream = admin.getInotifyEventStream();
        //转换时间格式---> 时间戳转为时间
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm");
        while (true){
            /*
                手动获取一批新的事件，如果不能立即获取到，则会无限期的卡在这里，一直等待,该功能也是避免死循环导致内存泄露的一种手段。
                take()方法官方解释原文：Returns the next batch of events in the stream, waiting indefinitely if
   * a new batch  is not immediately available.
             */
            EventBatch events = inotifyEventStream.take();
            for (Event event: events.getEvents()){
                //事件类型
//                System.out.println("事件类型是：" + event.getEventType());
                switch (event.getEventType()){
                    //1、创建事件：如果是创建目录的话，直接创建成功；如果是创建文件，先要创建一个后缀为._COPYING_的临时文件，之后再进行rename操作
                    //WebHDFS的上传文件不会生成中间文件，直接就是最终文件
                    case CREATE:
                        Event.CreateEvent create = (Event.CreateEvent) event;
                        String createPath = create.getPath();
//                        System.out.println("软链接是：" + create.getSymlinkTarget());
                        //todo 添加要过滤的目录前缀，控制所监控的目录
                        //这里监控test02目录
                        if (createPath.startsWith("/test02")){
                            //打印上传文件的信息、新建文件夹的信息。只有文件大小在close那一步体现
                            //todo 最近根据是否是文件夹，给权限前加d
                            System.out.println("文件：" + create.getPath().replace("._COPYING_", "") +
                                    "，权限：" + create.getPerms() +
                                    "，副本数：" + create.getReplication() +
                                    "，所属者：" + create.getOwnerName() +
                                    "，组名：" + create.getGroupName() +
                                    "，创建时间：" + simpleDateFormat.format(create.getCtime())
                            );
                            //获取本事件操作的INode的类型（三种类型）：file/directory/SYSLINK
                            Event.CreateEvent.INodeType iNodeType = create.getiNodeType();
                            switch (iNodeType){
                                case FILE:
                                    if (!createPath.endsWith("_COPYING_")){
                                        System.out.println("正在使用Restful风格上传文件，路径为：" + createPath);
                                    }
                                    break;
                                case DIRECTORY:
                                    if (createPath.startsWith("/test02")){
                                        System.out.println("创建目录成功，路径为：" + createPath);
                                    }
                                    break;
                                case SYMLINK:
                                    System.out.println("正在创建软链接，路径为：" + createPath);
                                    break;
                                default:
                                    break;
                            }
                        }
                        break;

                    //2、一是创建文件（生成最终文件）；二是删除文件；三是更改文件（夹）；四是移动文件（夹）
                    //删除文件夹，只提示删除文件夹，不显示删除文件
                    //WebHDFS的删除不会进回收站
                    case RENAME:
                        Event.RenameEvent rename = (Event.RenameEvent) event;
                        //源路径
                        String srcPath = rename.getSrcPath();
                        //目标路径
                        String dstPath = rename.getDstPath();
                        //rename时间
                        String renameTime = simpleDateFormat.format(rename.getTimestamp());
                        if (srcPath.startsWith(dstPath)&srcPath.endsWith("_COPYING_")&dstPath.startsWith("/test02")){
                            //如果源地址是以目的地址开头的，则可以判断是临时文件。当其rename之后，就变成了最终文件
                            System.out.println("正在使用非restful风格上传文件，路径为：" + dstPath);
                            //如果源地址的路径是以过滤路径开头，目的地址是以垃圾桶路径开头。则认为是删除目录（不跳过垃圾桶）
                        }else if (dstPath.startsWith("/user")&srcPath.startsWith("/test02")){
                            System.out.println("正在使用非restful风格删除文件，路径为：" + srcPath);
                            //剩下的情况就是修改文件（夹）名称、或者移动文件（夹）
                        }else {
                            if (dstPath.startsWith("/test02")&srcPath.startsWith("/test02")){
                                if (dstPath.split("/").length == srcPath.split("/").length) {
                                    System.out.println("正在更改文件（夹）的名称，原始路径为：" + srcPath);
                                    System.out.println("正在更改文件（夹）的名称，最终路径为：" + dstPath +
                                            "，更改时间：" + renameTime);
                                }else {
                                    System.out.println("正在移动文件（夹），原始路径为：" + srcPath);
                                    System.out.println("正在移动文件（夹），最终路径为：" + dstPath +
                                            "，移动时间：" + renameTime);
                                }
                            }
                        }
                        break;

                    //3、未知操作：比如跳过回收站删除文件（夹）
                    case UNLINK:
                        Event.UnlinkEvent unlinkEvent = (Event.UnlinkEvent) event;
                        String unlinkPath = unlinkEvent.getPath();
                        if (unlinkPath.startsWith("/test02")){
                            System.out.println("正在使用restful风格删除文件，路径为：" + unlinkPath);
                        }
                        break;

                    //4、当写完文件，rename之前，会进行这一步操作
                    //Sent when a file is closed after append or create.
                    case CLOSE:
                        Event.CloseEvent closeEvent = (Event.CloseEvent) event;
                        String closePath = closeEvent.getPath();
                        long fileSize = closeEvent.getFileSize();
                        //加过滤条件
                        if (closePath.startsWith("/test02")) {
                            System.out.println(closePath.replace("._COPYING_", "") +
                                    "文件大小为：" + fileSize);
                        }
                        break;

                    //5、给文件追加内容。追加之后的文件大小也会在close处输出
                    //shell命令追加为：-appendToFile
                    case APPEND:
                        Event.AppendEvent appendEvent = (Event.AppendEvent) event;
                        String appendPath = appendEvent.getPath();
                        System.out.println("正在使用shell命令行给文件追加内容，路径为：" + appendPath);
                        break;

                    case METADATA:
                        Event.MetadataUpdateEvent metadataUpdateEvent = (Event.MetadataUpdateEvent) event;
                        String metaPath = metadataUpdateEvent.getPath();
                        System.out.println("The METADATA path is：" + metaPath);
                        break;
                    default:
                        break;
                }
            }
        }
    }


}
